package org.cst.ming.android.barcode.comminucation;


/**
 * 
 */

import java.io.Closeable;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.Collections;
import java.util.Formatter;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.future.IoFutureListener;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.WriteToClosedSessionException;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.filter.logging.LogLevel;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.cst.ming.android.barcode.comminucation.SocketCommunicator.Message.MessageState;


/**
 * @author Ming
 * 
 */
public class SocketCommunicator
		implements Closeable
{
	private static enum ConnectState
	{
		READY,
		CONNECTING,
		ESTABLISHED,
		CLOSED
	}

	/** The SLF4J Logger. */
	private static final String TAG = "SocketCommunicator";
	private static final org.slf4j.Logger LOGGER = org.slf4j.LoggerFactory.getLogger(TAG);
	private static final LoggingFilter LOGGING_FILTER = new LoggingFilter(TAG);
	static
	{
		LOGGING_FILTER.setExceptionCaughtLogLevel(LogLevel.ERROR);
		LOGGING_FILTER.setSessionCreatedLogLevel(LogLevel.DEBUG);
		LOGGING_FILTER.setSessionOpenedLogLevel(LogLevel.DEBUG);
		LOGGING_FILTER.setSessionClosedLogLevel(LogLevel.DEBUG);
		LOGGING_FILTER.setMessageSentLogLevel(LogLevel.DEBUG);
		LOGGING_FILTER.setMessageReceivedLogLevel(LogLevel.DEBUG);
		LOGGING_FILTER.setSessionIdleLogLevel(LogLevel.DEBUG);
	}

	private final Map<String, Message<String>> history = Collections.synchronizedMap(new LinkedHashMap<String, Message<String>>());
	// private final Queue<String> waitings = new ConcurrentLinkedQueue<String>();
	private final BlockingQueue<String> sendQueue = new LinkedBlockingQueue<String>();

	// private final String hostname;
	// private final int port;
	private final InetSocketAddress remoteAddress;
	private NioSocketConnector socketConnector;

	private Worker sendWorker;

	private volatile ConnectFuture connectFuture;
	// private volatile IoSession session = null;
	// private volatile boolean canRequestConnect = true;
	private volatile ConnectState status;

	public SocketCommunicator(String hostname, int port)
	{
		// this.hostname = hostname;
		// this.port = port;
		this.remoteAddress = new InetSocketAddress(hostname, port);
		this.status = ConnectState.READY;
		prepareConnector();
		connect();
	}

	private void prepareConnector()
	{
		socketConnector = new NioSocketConnector();
		socketConnector.setDefaultRemoteAddress(remoteAddress);
		socketConnector.setConnectTimeoutMillis(30000L); // wait 30s will cause timeout.
		socketConnector.getFilterChain().addLast("eventLogging", LOGGING_FILTER);
		socketConnector.getFilterChain().addLast("codec",
				new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName("UTF8"))));
		socketConnector.setHandler(new IoHandlerAdapter()
		{
			@Override
			public void sessionOpened(IoSession session)
			{
				status = ConnectState.ESTABLISHED;
				sendWorker = new Worker(session);
				sendWorker.start();
			}

			@Override
			public void sessionClosed(IoSession session)
			{
				status = ConnectState.CLOSED;
				sendWorker.interrupt();
			}

			@Override
			public void exceptionCaught(IoSession session, Throwable cause)
			{
				status = ConnectState.CLOSED;
				LOGGER.error("Exception caught - [{}]:[{}]", cause.getClass().getName(), cause.getMessage());
			}
		});
	}

	/**
	 * Connect to the remote host. This method will call by constructor usually, call this method manually for reconnect
	 * while network is unstable.
	 */
	public final ConnectFuture connect()
	{
		if (ConnectState.READY != status && ConnectState.CLOSED != status)
		{
			LOGGER.warn("Should not connect while connecting or connected, ignoring operation");
			return connectFuture;
		}

		if (socketConnector.isDisposed() || socketConnector.isDisposing())
		{ // connector disposed, prepare a new one.
			prepareConnector();
		}

		status = ConnectState.CONNECTING;
		connectFuture = socketConnector.connect();
		return connectFuture;
	}

	// /**
	// * Wait the connection established or timeout. The method will block till the connection established or error
	// * occurs(i.e timeout).
	// *
	// * @return this instance.
	// * @throws RuntimeIoException If cannot establish connection to remote address.
	// */
	// public final SocketCommunicator waitConnect()
	// {
	// if (LOGGER.isDebugEnabled())
	// {
	// final double waitStart = System.nanoTime();
	// connectFuture.awaitUninterruptibly();
	// final double waitEnd = System.nanoTime();
	// final double elapsed = (waitEnd - waitStart) / (1000 * 1000);
	// LOGGER.debug("Waiting connect elapsed: {}ms", elapsed);
	// } else
	// {
	// connectFuture.awaitUninterruptibly();
	// }
	//
	// if (!connectFuture.isConnected() || null != connectFuture.getException())
	// {
	// canRequestConnect = true;
	// LOGGER.error("Cannot connect to remote host [{}]", remoteAddress);
	// throw new RuntimeIoException("Cannot connect to remote host");
	// }
	//
	// if (null == session)
	// {
	// session = connectFuture.getSession();
	// }
	// return this;
	// }

	/**
	 * Send the message out.
	 * 
	 * @param message The message to be sent.
	 * @throws IllegalStateException If connection is not established or closed.
	 */
	public final void send(final String message)
	{
		final Message<String> messageWrapper;
		if (!history.containsKey(message))
		{
			messageWrapper = new Message<String>(message);
			history.put(message, messageWrapper);
		} else
		{
			messageWrapper = history.get(message);
		}

		sendQueue.add(message);
		LOGGER.debug("Message has entered queue: [{}]", message);
	}

	public final Message<String> getMessage(String target)
	{
		final Message<String> messageWrapper = history.get(target);
		if (null == messageWrapper)
		{
			return null;
		}
		return messageWrapper;
	}

	// /**
	// * Get that target status or <code>null</code> if target not exists in history.
	// *
	// * @param target The target which is being queried.
	// * @return <code>null</code> if target not exists.
	// */
	// public final MessageState getStatus(String target)
	// {
	// final Message<String> messageWrapper = history.get(target);
	// if (null == messageWrapper)
	// {
	// return null;
	// }
	// return messageWrapper.getStatus();
	// }

	/**
	 * Get the history {@link Map}, <code><b><i>key</i></b></code> is the target which is being queried,
	 * <code><b><i>value</i></b></code> is the target and its additional info.
	 * 
	 * @return The unmodifiable Map.
	 */
	public final Map<String, Message<String>> getHistory()
	{
		return Collections.unmodifiableMap(history);
	}

	private final class Worker
			extends Thread
	{
		private final IoSession session;

		private Worker(IoSession session)
		{
			super("worker-thread");
			this.session = session;
		}

		@Override
		public void run()
		{
			while (true)
			{
				if (isInterrupted() || session.isClosing())
				{
					LOGGER.debug("Worker is exiting, because session close.");
					break;
				}
				try
				{
					run0();
				} catch (InterruptedException e)
				{
					LOGGER.debug("Worker is exiting, because session close.");
					break;
				}
			}
		}

		private void run0()
				throws InterruptedException
		{
			final String message = sendQueue.take();
			final Message<String> messageWrapper = history.get(message);
			messageWrapper.setStatus(MessageState.SENDING);
			final WriteFuture writeFuture = session.write(message).addListener(new IoFutureListener<WriteFuture>()
			{
				public void operationComplete(WriteFuture future)
				{
					// here future.isDone must be true.
					if (future.isWritten())
					{
						LOGGER.debug("Message sent: [{}]", message);
						messageWrapper.setStatus(MessageState.SENT);
					} else if (future.getException() instanceof WriteToClosedSessionException)
					{
						LOGGER.warn("Session closed, retry sending message: [{}]", message);
						send(message);
					} else
					{
						LOGGER.warn("Message cannot send: [{}], cause: [{}]", message, future.getException());
						messageWrapper.setStatus(MessageState.ERROR);
					}
				}
			});
			messageWrapper.setWriteFuture(writeFuture);
		}
	}

	/**
	 * The wrapper that wrap the message and any other additional informations.
	 * 
	 * @author Ming
	 * 
	 * @param <T> The actually message type.
	 */
	public static class Message<T>
	{

		public static enum MessageState
		{
			READY,
			SENDING,
			SENT,
			CONFIRMED,
			ERROR
		}

		private final T message;
		private volatile MessageState status;
		private volatile WriteFuture writeFuture;

		public Message(T message)
		{
			this.message = message;
			this.status = MessageState.READY;
		}

		public final T getContent()
		{
			return this.message;
		}

		public final MessageState getStatus()
		{
			if (this.status == MessageState.SENDING && writeFuture == null)
			{
				return MessageState.READY;
			}
			return this.status;
		}

		public final WriteFuture getWriteFuture()
		{
			return this.writeFuture;
		}

		@Override
		public String toString()
		{
			return new Formatter().format("[%s, %s]", message, status).toString();
		}

		private void setStatus(MessageState status)
		{
			this.status = status;
		}

		private void setWriteFuture(WriteFuture writeFuture)
		{
			this.writeFuture = writeFuture;
		}
	}

	public final void close()
	{
		close(false);
	}

	public final void close(boolean wait)
	{
		socketConnector.dispose(wait);
	}
}
